/*
 ***************************************************************************************
 * All rights Reserved, Designed By RTF
 * @Title:  RocketMqConsumerConfiger.java   
 * @Package com.rtf.repair.mq.rocketmq.consumer
 * @Description: 创建、销毁消费者   
 * @author: 司福林
 * @date:   2020-10-17 18:53:27   
 * @version V1.0 
 * @Copyright: 2020 RTF. All rights reserved.
 * 注意：本内容仅限于公司内部使用，禁止外泄以及用于其他的商业目
 *  ---------------------------------------------------------------------------------- 
 * 文件修改记录
 *     文件版本：         修改人：             修改原因：
 ***************************************************************************************
 */
package com.rtf.framework.mq.rocketmq.consumer;

import java.util.List;

import org.apache.commons.collections.CollectionUtils;
import org.springframework.util.Assert;

import com.rtf.framework.mq.common.CommonConstants;
import com.rtf.framework.mq.common.ConsumerProperties;
import com.rtf.framework.mq.common.Service;
import com.google.common.collect.Lists;

import lombok.extern.slf4j.Slf4j;

/**
 *  
 * @Description: 创建、销毁消费者
 * @ClassName:  RocketMqConsumerConfiger
 * @author: sfl
 * @date:   2020-10-17 18:56:01
 * @since:  v1.0
 */
@Slf4j
public class RocketMqConsumerConfiger implements Service {

	/** The properties. */
	private RocketMqConsumerProperties properties;

	/** The consumer properties. */
	private List<ConsumerProperties> consumerProperties;

	/** The consumers. */
	private List<RocketmqConsumerAdaptor> consumers = Lists.newArrayList();

	/**
	 * Instantiates a new rocket mq consumer configer.
	 *
	 * @param properties the properties
	 * @param cosumer the cosumer
	 */
	public RocketMqConsumerConfiger(RocketMqConsumerProperties properties, List<ConsumerProperties> cosumer) {

		Assert.notNull(properties, "RocketMqConsumerProperties is null");

		Assert.notEmpty(cosumer, "RocketMq consumer is empty");
		this.consumerProperties = cosumer;
		this.properties = properties;
	}

	/**
	 * Inits the.
	 */
	@Override
	public void init() {
		// TODO Auto-generated method stub

	}

	/**
	 * 根据ConsumerProperties.count 创建消费者
	 */
	@Override
	public void start() {

		log.info("##############begin create rocketmq consumer....#############");
		RocketmqConsumerAdaptor adaptor;
		String instanceName;
		for (ConsumerProperties consumer : this.consumerProperties) {
			instanceName = consumer.getInstanceName() + "_" + System.currentTimeMillis();
			if (CommonConstants.MQ_TYPE_ROCKETMQ.equals(consumer.getType())) {
				for (int i = 0; i < consumer.getCount(); i++) {
					consumer.setInstanceName(instanceName + "_" + i);
					adaptor = new RocketmqConsumerAdaptor(this.properties, consumer);
					adaptor.init();
					adaptor.start();
					consumers.add(adaptor);
				}

			}
		}
		log.info("############## create rocketmq {} consumer end #############", consumers.size());

	}

	/**
	 * Destroy.
	 */
	@Override
	public void destroy() {
		if (CollectionUtils.isNotEmpty(this.consumers)) {
			for (RocketmqConsumerAdaptor adaptor : this.consumers) {
				adaptor.destroy();
			}
		}

	}

}
